package promote.brozen.reactor;

import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.Fuseable;
import reactor.core.publisher.Operators;
import reactor.util.context.Context;

import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;


@Slf4j
public class BSubscriber implements CoreSubscriber<String>, Fuseable.QueueSubscription<String> {

    final CoreSubscriber<? super String> actual;

    Subscription s;
    Fuseable.QueueSubscription<String> qs;

    int streamId;
    boolean firstRequest = true;

    volatile int wip;
    static final AtomicIntegerFieldUpdater<BSubscriber> WIP =
            AtomicIntegerFieldUpdater.newUpdater(BSubscriber.class, "wip");

    public BSubscriber(CoreSubscriber<? super String> actual) {
        this.actual = actual;
    }

    @Override
    public Context currentContext() {
        return actual.currentContext();
    }

    @Override
    public void request(long n) {
        this.s.request(n);
        if (!firstRequest) {
            log.info("first request");
            return;
        }
        this.firstRequest = false;

        if (WIP.getAndIncrement(this) != 0) {
            return;
        }
        int missed = 1;

        boolean firstLoop = true;
        for (; ; ) {
            if (firstLoop) {
                firstLoop = false;
                log.info("first request");
            } else {
                log.info("cancel");
                return;
            }

            missed = WIP.addAndGet(this, -missed);
            if (missed == 0) {
                return;
            }
        }
    }

    @Override
    public void cancel() {
        this.s.cancel();

        if (WIP.getAndIncrement(this) != 0) {
            return;
        }

        log.info("cancel");
    }

    @Override
    @SuppressWarnings("unchecked")
    public void onSubscribe(Subscription s) {
        if (Operators.validate(this.s, s)) {
            this.s = s;
            if (s instanceof Fuseable.QueueSubscription) {
                this.qs = (Fuseable.QueueSubscription<String>) s;
            }
            this.actual.onSubscribe(this);
        }
    }

    @Override
    public void onNext(String t) {
        this.actual.onNext(t);
    }

    @Override
    public void onError(Throwable t) {
        this.actual.onError(t);
        try {
            log.info("terminal error");
        } catch (Throwable throwable) {
            Operators.onErrorDropped(throwable, currentContext());
        }
    }

    @Override
    public void onComplete() {
        this.actual.onComplete();
        try {
            log.info("terminal complete");
        } catch (Throwable throwable) {
            Operators.onErrorDropped(throwable, currentContext());
        }
    }

    @Override
    public int requestFusion(int requestedMode) {
        if (this.qs != null) {
            return this.qs.requestFusion(requestedMode);
        } else {
            return Fuseable.NONE;
        }
    }

    @Override
    public String poll() {
        return this.qs.poll();
    }

    @Override
    public int size() {
        return this.qs.size();
    }

    @Override
    public boolean isEmpty() {
        return this.qs.isEmpty();
    }

    @Override
    public void clear() {
        this.qs.clear();
    }
}
